package com.blb.edusearchservice.listener;

import com.alibaba.fastjson.JSON;
import com.blb.common.entity.Course;
import com.blb.edusearchservice.service.CourseIndexService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class CourseMQListener {

    //对课程进行添加或更新的队列名
    public static final String QUEUE_COURSE_SAVE = "queue.course.save";
    //对课程进行删除的队列名
    public static final String QUEUE_COURSE_REMOVE = "queue.course.remove";
    //对课程进行添加或更新的路由键
    public static final String KEY_COURSE_SAVE = "key.course.save";
    //对课程进行删除的路由键
    public static final String KEY_COURSE_REMOVE = "key.course.remove";
    //课程交换机名
    public static final String COURSE_EXCHANGE = "edu.course.exchange";

    @Autowired
    private CourseIndexService courseIndexService;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 监听课程添加操作
     */
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = QUEUE_COURSE_SAVE, durable = "true"),
                    exchange = @Exchange(value = COURSE_EXCHANGE,type = ExchangeTypes.TOPIC,ignoreDeclarationExceptions = "true")
                    , key = KEY_COURSE_SAVE)})
    public void receiveCourseSaveMessage(String data, Channel channel, Message message) {
        try {
            log.info("课程保存：{}",data);
            //获得消息的id
            String id = message.getMessageProperties().getHeader("spring_returned_message_correlation");
            //将id保存到redis中，如果id存在就不执行业务逻辑
            ValueOperations<String,String> ops = redisTemplate.opsForValue();
            //如果id不存在就,设置id为键，执行业务逻辑
            if(ops.setIfAbsent(id,"0",100, TimeUnit.SECONDS)){
                //将json转换为课程对象
                Course course = JSON.parseObject(data, Course.class);
                courseIndexService.saveCourse(course);
                //逻辑执行完将消息状态设置为1
                ops.set(id,"1",100, TimeUnit.SECONDS);
                //手动确认
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                log.info("消息{}执行完成",id);
            }else{
                //id存在，重复消息
                log.info("消息已经存在，重复消息 {}",id);
                if("1".equals(ops.get(id))){
                    log.info("该消息已经执行完毕 {}",id);
                    //手动确认
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                }
            }
        } catch (Exception ex) {
            log.error("接收消息出现异常",ex);
        }
    }

    /**
     * 监听课程删除操作
     */
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = QUEUE_COURSE_REMOVE, durable = "true"),
                    exchange = @Exchange(value = COURSE_EXCHANGE,type = ExchangeTypes.TOPIC,ignoreDeclarationExceptions = "true"),
                    key = KEY_COURSE_REMOVE)})
    public void receiveCourseDeleteMessage(Long id, Channel channel, Message message) {
        try {
            log.info("课程删除完成：{}",id);
            courseIndexService.removeCourse(String.valueOf(id));
            //手动确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception ex) {
            log.error("接收消息出现异常",ex);
        }
    }
}
